package cn.mesmile.flink.transform;

import cn.mesmile.flink.jdkstream.VideoOrder;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Date;

/**
 * @author zb
 * @date 2021/8/22 23:14
 * @Description   keyBy 、 maxBy
 */
public class FlinkTransformKeyByMaxByApp {

    /**
     *  如果是用了keyby,在后续算子要用maxby，minby类型，才可以再分组里面找对应的数据
     *
     *      如果是用max、min等，就不确定是哪个key中选了
     *          如果是keyBy的是对象的某个属性，则分组用【max/min】聚合统计，【只有聚合的字段会更新】，【其他字段还是旧的】，【导致对象不准确】
     *         【需要】用【maxby/minBy】才对让整个【对象的属性都是最新的】
     *
     *
     *  常规的数据流转
     *      DataStream->keyBy操作->KeyStream->window操作->windowStream->聚合操作->DataStream
     *      注意: KeyBy后的聚合函数，只处理当前分组后组内的数据，不同组内数据互不影响
     */

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);

        DataStream<VideoOrder> ds = env.fromElements(
                new VideoOrder("21312","java",32,5,new Date()),
                new VideoOrder("314","java",32,5,new Date()),
                new VideoOrder("542","springboot",32,5,new Date()),
                new VideoOrder("42","redis",42,5,new Date()),
                new VideoOrder("52","java",32,5,new Date()),
                new VideoOrder("55","test",52,5,new Date()),
                new VideoOrder("56","cloud",62,5,new Date()),
                new VideoOrder("523","redis",42,5,new Date())
        );

        KeyedStream<VideoOrder, String> videoOrderStringKeyedStream = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder videoOrder) throws Exception {
                return videoOrder.getTitle();
            }
        });

        /**
         *  如果是keyBy的是对象的某个属性，则分组用【max/min】聚合统计，【只有聚合的字段会更新】，【其他字段还是旧的】，【导致对象不准确】
         * 【需要】用【maxby/minBy】才对让整个【对象的属性都是最新的】
         */
        SingleOutputStreamOperator<VideoOrder> money1 = videoOrderStringKeyedStream.maxBy("money");

//        SingleOutputStreamOperator<VideoOrder> money = videoOrderStringKeyedStream.sum("money");

        money1.print("result:");

        env.execute("custom keyBy app");
    }
}
